// Copyright (C) Kumo inc. and its affiliates.
// Author: Jeff.li lijippy@163.com
// All rights reserved.
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published
// by the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program.  If not, see <https://www.gnu.org/licenses/>.
//


#include <cmath>

#include <double-conversion/double-conversion.h>
#include <melon/expected.h>

#include <pollux/expression/presto_cast_hooks.h>
#include <pollux/common/strings/string_impl.h>
#include <pollux/type/timestamp_conversion.h>
#include <pollux/type/tz/time_zone_map.h>

namespace kumo::pollux::exec {
    PrestoCastHooks::PrestoCastHooks(const core::QueryConfig &config)
        : CastHooks(), legacyCast_(config.isLegacyCast()) {
        if (!legacyCast_) {
            options_.zeroPaddingYear = true;
            options_.dateTimeSeparator = ' ';
            const auto sessionTzName = config.sessionTimezone();
            if (config.adjustTimestampToTimezone() && !sessionTzName.empty()) {
                options_.timeZone = tz::locateZone(sessionTzName);
            }
        }
    }

    Expected<Timestamp> PrestoCastHooks::castStringToTimestamp(
        const StringView &view) const {
        const auto conversionResult = util::fromTimestampWithTimezoneString(
            view.data(),
            view.size(),
            legacyCast_
                ? util::TimestampParseMode::kLegacyCast
                : util::TimestampParseMode::kPrestoCast);
        if (conversionResult.hasError()) {
            return melon::makeUnexpected(conversionResult.error());
        }

        return util::fromParsedTimestampWithTimeZone(
            conversionResult.value(), options_.timeZone);
    }

    Expected<Timestamp> PrestoCastHooks::castIntToTimestamp(int64_t seconds) const {
        return melon::makeUnexpected(
            Status::UserError("Conversion to Timestamp is not supported"));
    }

    Expected<std::optional<Timestamp> > PrestoCastHooks::castDoubleToTimestamp(
        double seconds) const {
        return melon::makeUnexpected(
            Status::UserError("Conversion to Timestamp is not supported"));
    }

    Expected<int32_t> PrestoCastHooks::castStringToDate(
        const StringView &dateString) const {
        // Cast from string to date allows only complete ISO 8601 formatted strings:
        // [+-](YYYY-MM-DD).
        return util::fromDateString(dateString, util::ParseMode::kPrestoCast);
    }

    namespace {
        using double_conversion::StringToDoubleConverter;

        template<typename T>
        Expected<T> doCastToFloatingPoint(const StringView &data) {
            static const T kNan = std::numeric_limits<T>::quiet_NaN();
            static StringToDoubleConverter stringToDoubleConverter{
                StringToDoubleConverter::ALLOW_TRAILING_SPACES,
                /*empty_string_value*/ kNan,
                /*junk_string_value*/ kNan,
                "Infinity",
                "NaN"
            };
            int processedCharactersCount;
            T result;
            auto *begin = std::find_if_not(data.begin(), data.end(), [](char c) {
                return functions::stringImpl::isAsciiWhiteSpace(c);
            });
            auto length = data.end() - begin;
            if (length == 0) {
                // 'data' only contains white spaces.
                return melon::makeUnexpected(Status::UserError());
            }
            if constexpr (std::is_same_v<T, float>) {
                result = stringToDoubleConverter.StringToFloat(
                    begin, length, &processedCharactersCount);
            } else if constexpr (std::is_same_v<T, double>) {
                result = stringToDoubleConverter.StringToDouble(
                    begin, length, &processedCharactersCount);
            }
            // Since we already removed leading space, if processedCharactersCount == 0,
            // it means the remaining string is either empty or a junk string. So return a
            // user error in this case.
            if UNLIKELY(processedCharactersCount == 0) {
                return melon::makeUnexpected(Status::UserError());
            }
            return result;
        }
    } // namespace

    Expected<float> PrestoCastHooks::castStringToReal(
        const StringView &data) const {
        return doCastToFloatingPoint<float>(data);
    }

    Expected<double> PrestoCastHooks::castStringToDouble(
        const StringView &data) const {
        return doCastToFloatingPoint<double>(data);
    }

    StringView PrestoCastHooks::removeWhiteSpaces(const StringView &view) const {
        return view;
    }

    const TimestampToStringOptions &PrestoCastHooks::timestampToStringOptions()
    const {
        return options_;
    }

    PolicyType PrestoCastHooks::getPolicy() const {
        return legacyCast_
                   ? PolicyType::LegacyCastPolicy
                   : PolicyType::PrestoCastPolicy;
    }
} // namespace kumo::pollux::exec
